package cn.tedu.topic;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;

public class TopicConsumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂对象
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("192.168.8.100");
        cf.setPort(5672);   //默认端口号，可以省略不写
        cf.setUsername("test");
        cf.setPassword("test");
        cf.setVirtualHost("/");  //如果不指定虚拟主机，默认是"/",可以省略不写
        //2.通过连接工厂对象创建tcp连接对象，通过tcp连接对象创建channel
        Connection nc = cf.newConnection();
        Channel cc = nc.createChannel();
        //3.通过channel创建交换机
        //第一个参数：交换机的名称
        //第二个参数：交换机的类型
        cc.exchangeDeclare("topicMode.logs", BuiltinExchangeType.TOPIC);
        //4.通过channel创建队列,空构造可以创建一个随机队列名，非持久、独占、自动删除的队列
        String queueName = cc.queueDeclare().getQueue();
        //5.绑定队列和交换机
        System.out.print("请输入绑定键，多个绑定键之间用空格隔开:");
        String str = new Scanner(System.in).nextLine();
        String[] bindingKeys = str.split("\\s+");
        for (String bindingKey : bindingKeys) {
            //队列名   交换机名    绑定键
            cc.queueBind(queueName, "topicMode.logs", bindingKey);
        }

        //6.消费数据
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                //delivery里面封装了要处理的消息
                byte[] body = delivery.getBody();
                //转成对应数据类型的数据进行处理
                String msg = new String(body);
                String routingKey = delivery.getEnvelope().getRoutingKey();
                System.out.println("要处理" + msg + "消息数据了");
                System.out.println(msg + "处理完毕" + routingKey);
                System.out.println("================================");
            }
        };
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
            }
        };
        cc.basicConsume(queueName, true, deliverCallback, cancelCallback);
    }
}
